package com.atguigu.Flink.process;

import com.alibaba.fastjson.JSON;
import com.atguigu.Flink.POJO.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Flink02_TimeService {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        ds.print("input");

        //测试定时器

        ds.keyBy(
                k->k.getUser()
        ).process(
                new KeyedProcessFunction<String, Event, String>() {
                    //处理每条数据
                    @Override
                    public void processElement(Event event, KeyedProcessFunction<String, Event, String>.Context context, Collector<String> collector) throws Exception {
//                        long l = context.timerService().currentProcessingTime();
//                        System.out.println("当前的系统时间为"+l);
//                        long timerTime = l + 5000L;
//                        context.timerService().registerProcessingTimeTimer(timerTime);
//                        System.out.println("注册系统时间定时器" + timerTime);
//                        collector.collect(JSON.toJSONString(event));

                        //测试事件时间的定时器
                        long l = context.timerService().currentWatermark();
                        System.out.println("当前的水位线："+l);
                        long timerTime = l + 5000L;
                        context.timerService().registerEventTimeTimer(timerTime);
                        System.out.println("注册事件时间定时器"+timerTime);
                        collector.collect(JSON.toJSONString(event));

                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
//                        System.out.println("dangqianchufadedingshiqiwei:"+ timestamp);
                        System.out.println("当前触发的定时器为"+timestamp);
                    }
                }
        ).print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
